/*
 * Copyright 2009-2010 by The Regents of the University of California
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * you may obtain a copy of the License from
 * 
 *     http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package edu.ucr.cs.dblab.acestore.operators;

import java.nio.ByteBuffer;

import edu.ucr.cs.dblab.acestore.api.IOperator;
import edu.ucr.cs.dblab.acestore.api.ISinglePrevOperatorFactory;
import edu.ucr.cs.dblab.acestore.api.PageAccessHelper;
import edu.ucr.cs.dblab.acestore.api.TupleInterpreter;
import edu.ucr.cs.dblab.acestore.base.ACEContext;

/**
 * @author jarodwen
 * @created Jun 3, 2011
 */
public class PagePrintOepratorFactory implements ISinglePrevOperatorFactory {

    /* (non-Javadoc)
     * @see edu.ucr.cs.dblab.acestore.api.ISinglePrevOperatorFactory#createOperator(edu.ucr.cs.dblab.acestore.base.ACEContext, edu.ucr.cs.dblab.acestore.api.IOperator)
     */
    @Override
    public IOperator createOperator(final ACEContext ctx, final IOperator prevOperator) {
        return new IOperator() {
            
            @Override
            public void open() throws Exception {
                prevOperator.open();
            }
            
            @Override
            public void getNext(ByteBuffer page) throws Exception {
                page.position(0);
                int nextPageID = 0;
                while(nextPageID >= 0){
                    prevOperator.getNext(page);
                    System.out.println("--- Page " + nextPageID);
                    page.position(0);
                    int recCount = PageAccessHelper.GetPageRecCount(page);
                    for(int i = 0; i < recCount; i++){
                        int startOffset = PageAccessHelper.GetRecStartOffset(page, i);
                        int length = PageAccessHelper.GetRecEndOffset(page, i) - startOffset + 1;
                        TupleInterpreter tupleInterpreter = new TupleInterpreter(page.array(), startOffset, length);
                        for(int j = 0; j < tupleInterpreter.getFieldCount(); j++){
                            System.out.print(getByteString(tupleInterpreter.getField(j)) + ",");
                        }
                        System.out.println();
                    }
                    nextPageID = PageAccessHelper.GetPageReference(page);
                    
                }
            }
            
            private String getByteString(byte[] array){
                StringBuilder sbder = new StringBuilder();
                sbder.append("[");
                for(int i = 0; i < array.length; i++){
                    sbder.append(array[i]);
                    if(i < array.length - 1)
                        sbder.append(",");
                }
                sbder.append("]");
                return sbder.toString();
            }
            
            @Override
            public void close() throws Exception {
                // TODO Auto-generated method stub
                
            }
        };
    }

    /**
     * @param args
     */
    public static void main(String[] args) {
        // TODO Auto-generated method stub

    }

}
